/**
 * 事件定时器
 */


package com.atguigu.day4.keyed

import java.sql.Timestamp

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object EventTimeTimer {
  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //    设置时间语义为事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    env.setParallelism(1)
    //
    val stream1 = env.socketTextStream("192.168.35.102", 9999, '\n')
      .map(line => {
        var arr = line.split(" ")
        (arr(0), arr(1).toLong * 1000L)
      })
      .assignAscendingTimestamps(_._2)
      .keyBy(_._1)
      .process(new Keyed)
      .print()
  env.execute()
  }

  class Keyed extends KeyedProcessFunction[String,(String,Long),String]{
    override def processElement(value: (String, Long), ctx: KeyedProcessFunction[String, (String, Long), String]#Context, out: Collector[String]): Unit = {
     //注册一个定时器：时间携带时间戳加10秒
      ctx.timerService().registerEventTimeTimer(value._2+10*1000L)
    }
    //定时器触发时执行内容
    override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, (String, Long), String]#OnTimerContext, out: Collector[String]): Unit = {
      out.collect("时间戳为"+ new Timestamp(timestamp))
    }
  }
}